package com.giw.project.advanced_graph_traversal;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text; //import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.giw.project.exception_manager.UncheckedException;

public class TraversalApp extends Configured implements Tool {

	private final static String tmp_path = "tmp.txt";
	private final static String tmp_path2 = "tmp2.txt";
	private final static String tmp_pathS = "tmp_searchKind.txt";

	private static Integer genreLimit = 50;
	private static Integer actorLimit = 50;
	private static Integer directorLimit = 50;

	private int genre_rank;
	private int actor_rank;
	private int director_rank;

	private Map<String, Integer> movie2rank = new HashMap<String, Integer>();

	private final static String absolute_path = "/home/giorgio/Scrivania/GIW/progetto/0-grafo/Split/newSplit/";

	private final static String tmp_endSearch = "tmp_endSearch.txt";

	private final static Map<String, String> node2input = new HashMap<String, String>();
	private String uriOriginalMovie;

	private void initializeMap() {
		node2input.put("string", absolute_path + "litteral");
		node2input.put("<http://data.linkedmdb.org/resource/film/",
				absolute_path + "movie");
		node2input.put("<http://data.linkedmdb.org/resource/director/",
				absolute_path + "director");
		node2input.put("<http://data.linkedmdb.org/resource/film_genre/",
				absolute_path + "genre");
		node2input.put("<http://data.linkedmdb.org/resource/actor/",
				absolute_path + "actor");
		node2input.put("else", absolute_path + "else");

	}

	public int run(String[] args) throws Exception {
		this.initializeMap();
		boolean success = true;
		int count_iteration = 0;

		FileOutputStream file = new FileOutputStream(tmp_endSearch);
		file.close();

		File fS = new File(tmp_pathS);
		FileOutputStream fosS = new FileOutputStream(fS);
		PrintStream outputS = new PrintStream(fosS);
		int i = 2;
		for (i = 2; i < 5; i++) {
			outputS.println(args[i]);

		}
		outputS.close();
		fosS.close();

		int directorWeight = Integer.parseInt(args[2]);
		int genreWeight = Integer.parseInt(args[3]);
		int actorWeight = Integer.parseInt(args[4]);

		int weightsSum = directorWeight + genreWeight + actorWeight;

		genreLimit *= genreWeight;
		actorLimit *= actorWeight;
		directorLimit *= directorWeight;
		int rimanenti = 500 - (genreLimit + actorLimit + directorLimit);
		if (rimanenti > 0) {
			int subparts = rimanenti / weightsSum;
			genreLimit += subparts * genreWeight;
			actorLimit += subparts * actorWeight;
			directorLimit += subparts * directorWeight;

		}

		File f = new File(tmp_path);
		FileOutputStream fos = new FileOutputStream(f);
		PrintStream output = new PrintStream(fos);
		if (Integer.parseInt(args[2]) != 0)
			output.println(args[1].toLowerCase() + "\t"
					+ "<http://purl.org/dc/terms/title>" + "\t" + "director");
		if (Integer.parseInt(args[3]) != 0)
			output.println(args[1].toLowerCase() + "\t"
					+ "<http://purl.org/dc/terms/title>" + "\t" + "genre");
		if (Integer.parseInt(args[4]) != 0)
			output.println(args[1].toLowerCase() + "\t"
					+ "<http://purl.org/dc/terms/title>" + "\t" + "actor");
		output.close();
		fos.close();
		List<String> inputs;

		File fileEnd = new File(tmp_endSearch);
		this.initializeRank();
		

		while (fileEnd.exists()) {
			count_iteration++;

			inputs = this.getInputDirectory();
			if (inputs == null) {
				throw new UncheckedException(
						"There's no an input for the that kind of node,"
								+ "see the tmp.txt file");

			}
			if (count_iteration == 2) {
				this.getUriOriginalMovie();
			}
			Job job = new Job(getConf());
			job.setJobName("Graph_Traversal");
			job.setJarByClass(TraversalMapper.class);
			job.setMapperClass(TraversalMapper.class);
			job.setReducerClass(TraversalReducer.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(Text.class);
			job.setInputFormatClass(TextInputFormat.class);
			job.setOutputFormatClass(TextOutputFormat.class);
			for (String input : inputs) {
				FileInputFormat.addInputPath(job, new Path(input));
			}
			FileOutputFormat.setOutputPath(job, new Path(args[0]
					+ count_iteration));

			success = job.waitForCompletion(true);
			f.delete();
			if (count_iteration == 3){
				this.initializeMovie2Rank();
				this.cleanFile();}
			File file1 = new File(tmp_path2);

			File file2 = new File(tmp_path);
			file1.renameTo(file2);

		}
		i = 1;
		for (i = 1; i < count_iteration; i++) {
			f = new File(args[0] + count_iteration);
			f.delete();
		}
		f = new File(tmp_path);
		f.delete();
		f = new File(tmp_pathS);
		f.delete();
		this.eliminateOriginalMovie(args[0] + count_iteration);
		this.optimazeResult(args[0] + count_iteration + "/part-r-00000");
		return success ? 0 : 1;
	}

	private void cleanFile() {
		File f = new File(tmp_path2);
		FileInputStream fis;
		String line = null;
		

		// List<String> SeenGenre = new LinkedList<String>();
		// List<String> SeenActor = new LinkedList<String>();
		// List<String> ToWrite = new LinkedList<String>();

		Map<Integer, List<String>> Genre = new TreeMap<Integer, List<String>>(new InverseIntegerOrder());
		Map<Integer, List<String>> Actor = new TreeMap<Integer, List<String>>(new InverseIntegerOrder());
		Map<Integer, List<String>> Director = new TreeMap<Integer, List<String>>(new InverseIntegerOrder());

		try {
			fis = new FileInputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException(e.getMessage());
		}
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		try {

			line = buffer.readLine();
			// int genreCount = 1;
			// int actorCount = 1;
			// int directorCount = 1;
			while (line != null) {
				String movie = line.split("\t")[0];

				String genre = line.split("\t")[2];
				if (genre.equals("genre")
				// && genreCount < genreLimit
						&& !Genre.values().contains(line)) {
					// if (!SeenGenre.contains(movie))
					// SeenGenre.add(movie);
					// else {
					// ToWrite.add(line);
					// ToWrite.add(line);
					// genreCount++;
					// }
					int tempRank = movie2rank.get(movie);
					List<String> temp = Genre.get(tempRank);
					if (temp == null)
						temp = new LinkedList<String>();
					temp.add(line);
					Genre.put(tempRank, temp);

				}
				if (genre.equals("actor")
				// && actorCount < actorLimit
						&& !Actor.values().contains(line)) {
					// if (!SeenActor.contains(movie))
					// SeenActor.add(movie);
					// else {
					// ToWrite.add(line);
					// ToWrite.add(line);
					// actorCount++;
					
					int tempRank = movie2rank.get(movie);
					List<String> temp = Actor.get(tempRank);
					if (temp == null)
						temp = new LinkedList<String>();
					temp.add(line);
					Actor.put(tempRank, temp);
				}

				if (genre.equals("director")
				// && directorCount < directorLimit
						&& !Director.values().contains(line)) {
					// ToWrite.add(line);
					// directorCount++;
					int tempRank = movie2rank.get(movie);
					List<String> temp = Director.get(tempRank);
					if (temp == null)
						temp = new LinkedList<String>();
					temp.add(line);
					Director.put(tempRank, temp);

				}

				line = buffer.readLine();
				// if (directorCount == directorLimit && actorCount ==
				// actorLimit
				// && genreCount == genreLimit)
				// break;

			}
		} catch (IOException e) {
			throw new UncheckedException(e.getMessage());
		}
		try {
			buffer.close();
			is.close();
			fis.close();
		} catch (IOException e) {
			e.printStackTrace();
		}

		f.delete();

		f = new File(tmp_path2);
		FileOutputStream fos;
		try {
			fos = new FileOutputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException(e.getMessage());
		}
		PrintStream output = new PrintStream(fos);

		int contatore = 0;

		for (Integer rank : Genre.keySet()) {
			if (contatore < genreLimit) {
				List<String> temp = new LinkedList<String>();
				temp = Genre.get(rank);
				for (String subparts : temp) {
					if (contatore < genreLimit) {
						for(int u = 0; u < rank/genre_rank ; u++){
						output.println(subparts);}
						contatore++;
					} else {
						break;
					}
				}
			} else {
				contatore = 0;
				break;
			}
		}

		for (Integer rank : Actor.keySet()) {
			if (contatore < actorLimit) {
				List<String> temp = new LinkedList<String>();
				temp = Actor.get(rank);
				for (String subparts : temp) {
					if (contatore < actorLimit) {
						for(int u = 0; u < rank/actor_rank ; u++){
							output.println(subparts);}
						contatore++;
					} else {
						break;
					}
				}
			} else {
				contatore = 0;
				break;
			}
		}

		for (Integer rank : Director.keySet()) {
			if (contatore < directorLimit) {
				List<String> temp = new LinkedList<String>();
				temp = Director.get(rank);
				for (String subparts : temp) {
					if (contatore < directorLimit) {
						for(int u = 0; u < rank/director_rank ; u++){
							output.println(subparts);}
						contatore++;
					} else {
						break;
					}
				}
			} else {
				contatore = 0;
				break;
			}
		}
		// for (String result : ToWrite) {
		// output.println(result);
		// }

		output.close();
		try {
			fos.close();
		} catch (IOException e) {
			throw new UncheckedException(e.getMessage());
		}

	}

	public static void main(String[] args) throws Exception {
		if (args.length != 5) {
			System.err
					.println("Usage: <output path> <movie-title> <director-rank> <genre-rank> <cast-rank>");
			// search kind = genre or director or actors
			System.exit(-1);
		}
		int ret = ToolRunner.run(new TraversalApp(), args);
		System.exit(ret);
	}

	public List<String> getInputDirectory() {
		File f = new File(tmp_path);
		FileInputStream fis;
		String line = null;
		List<String> inputs = new LinkedList<String>();
		try {
			fis = new FileInputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException("No input directory!\nMessage:"
					+ e.getMessage());
		}
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		try {
			line = buffer.readLine();
			while (line != null) {
				String kindNode = this.getNodeKind(line.split("\t")[0]);
				if (kindNode == null)
					kindNode = "else";
				String input = node2input.get(kindNode);
				String node = line.split("\t")[0];
				if (!node.contains("http")) {
					node = node.toLowerCase();
					if (node.charAt(0) == '"')
						node = node.substring(1, (node.length() - 1));
					while (node.charAt(0) == ' ')
						node = node.substring(1, (node.length() - 1));
					input += "/" + node.charAt(0);
				} else {
					if (!kindNode.equals("else")) {
						if (!kindNode
								.equals("<http://data.linkedmdb.org/resource/film/")) {
							String[] parts_node = node.split("/");
							int id = 0;
							String a = parts_node[(parts_node.length - 1)];
							a = a.substring(0, a.length() - 1);
							id = Integer.parseInt(a);
							id = id / 1000;
							input += "/" + id;
						} else {
							String[] parts_node = node.split("/");
							int id = 0;
							String a = parts_node[(parts_node.length - 1)];
							a = a.substring(0, a.length() - 1);
							id = Integer.parseInt(a);
							input += "/" + id;
						}

					}
				}
				if (!inputs.contains(input))
					inputs.add(input);
				line = buffer.readLine();
			}
		} catch (IOException e) {
			throw new UncheckedException("No line to read!\nMessage:"
					+ e.getMessage());
		}
		try {
			buffer.close();
			is.close();
			fis.close();
		} catch (IOException e) {
			e.printStackTrace();
		}

		return inputs;
	}

	private String getNodeKind(String uri) {
		if (!uri.contains("http")) {
			return "string";
		}
		String[] parts_uri = uri.split("/");
		int i;
		uri = "";
		for (i = 0; i < parts_uri.length - 1; i++) {
			uri += "/" + parts_uri[i];
		}
		return uri.substring(1) + "/";

	}

	private void getUriOriginalMovie() {
		File f = new File(tmp_path);
		FileInputStream fis;
		String line = null;
		try {
			fis = new FileInputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException(e.getMessage());
		}
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		try {
			line = buffer.readLine();
		} catch (IOException e) {
			throw new UncheckedException(e.getMessage());
		}
		try {
			buffer.close();
			is.close();
			fis.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		this.uriOriginalMovie = line.split("\t")[0];
	}

	private void eliminateOriginalMovie(String path) {
		List<String> results = new LinkedList<String>();
		File f = new File(path + "/part-r-00000");
		FileInputStream fis;
		String line = null;
		try {
			fis = new FileInputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException(e.getMessage());
		}
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		try {
			line = buffer.readLine();
			while (line != null) {
				if (!(line.split("\t")[0].equals(this.uriOriginalMovie))) {
					results.add(line);
				}
				line = buffer.readLine();
			}
		} catch (IOException e) {
			throw new UncheckedException(e.getMessage());
		}
		try {
			buffer.close();
			is.close();
			fis.close();
		} catch (IOException e) {
			e.printStackTrace();
		}

		f.delete();
		f = new File(path + "/part-r-00000");
		FileOutputStream fos;
		try {
			fos = new FileOutputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException(e.getMessage());
		}
		PrintStream output = new PrintStream(fos);
		for (String result : results) {
			output.println(result);
		}
		output.close();
		try {
			fos.close();
		} catch (IOException e) {
			throw new UncheckedException(e.getMessage());
		}
	}

	private void optimazeResult(String path) {
		Map<String, Integer> nodelink2rank = new HashMap<String, Integer>();
		File f = new File(path);
		FileInputStream fis;
		String line = null;
		try {
			fis = new FileInputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException(e.getMessage());
		}
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		try {
			line = buffer.readLine();
			while (line != null) {
				String node = line.split("\t")[0];
				String rank = line.split("\t")[1];
				String link = line.split("\t")[2];
				if (!nodelink2rank.containsKey(node + "\t" + link)) {
					nodelink2rank.put(node + "\t" + link, Integer
							.parseInt(rank));
				}
				line = buffer.readLine();
			}
		} catch (IOException e) {
			throw new UncheckedException(e.getMessage());
		}
		try {
			buffer.close();
			is.close();
			fis.close();
		} catch (IOException e) {
			e.printStackTrace();
		}

		f.delete();
		f = new File(path);
		FileOutputStream fos;
		try {
			fos = new FileOutputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException(e.getMessage());
		}
		PrintStream output = new PrintStream(fos);
		for (String result : nodelink2rank.keySet()) {
			output.println(result.split("\t")[0] + "\t"
					+ nodelink2rank.get(result) + "\t" + result.split("\t")[1]);
		}
		output.close();
		try {
			fos.close();
		} catch (IOException e) {
			throw new UncheckedException(e.getMessage());
		}

	}

	private Map<String, Integer> getMovie2Rank(String uri_link)
			throws IOException {
		File f = new File(tmp_path2);
		FileInputStream fis = new FileInputStream(f);
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		String line = buffer.readLine();
		Map<String, Integer> movie2rank = new HashMap<String, Integer>();
		while (line != null) {
			String currentNode = line.split("\t")[0];
			if (currentNode.equals(uri_link)) {
				String currentFlow = line.split("\t")[2];
				Integer current_rank = 0;
				if (currentFlow.equals("director"))
					current_rank = this.director_rank;
				if (currentFlow.equals("genre"))
					current_rank = this.genre_rank;
				if (currentFlow.equals("actor"))
					current_rank = this.actor_rank;
				Integer count = movie2rank.get(currentNode);
				if (count == null)
					count = 0;
				count += current_rank;
				movie2rank.put(currentNode, count);
			}
			line = buffer.readLine();
		}
		buffer.close();
		is.close();
		fis.close();
		return movie2rank;

	}

	private void initializeMovie2Rank() throws IOException {
		File f = new File(tmp_path2);
		FileInputStream fis = new FileInputStream(f);
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		String line = buffer.readLine();
		while (line != null) {
			String currentNode = line.split("\t")[0];
			{
				String currentFlow = line.split("\t")[2];
				Integer current_rank = 0;
				if (currentFlow.equals("director"))
					current_rank = this.director_rank;
				if (currentFlow.equals("genre"))
					current_rank = this.genre_rank;
				if (currentFlow.equals("actor"))
					current_rank = this.actor_rank;
				Integer count = movie2rank.get(currentNode);
				if (count == null)
					count = 0;
				count += current_rank;
				movie2rank.put(currentNode, count);
			}
			line = buffer.readLine();
		}
		buffer.close();
		is.close();
		fis.close();
		

	}

	private void initializeRank() throws IOException {
		File f = new File(tmp_pathS);
		FileInputStream fis = new FileInputStream(f);
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		String line = buffer.readLine();
		this.director_rank = Integer.parseInt(line);
		line = buffer.readLine();
		this.genre_rank = Integer.parseInt(line);
		line = buffer.readLine();
		this.actor_rank = Integer.parseInt(line);
		buffer.close();
		is.close();
		fis.close();
	}

}

// SP6nj7aB5mc5